package trigger

import (
	"bgs/db"
	sqlc "bgs/db/sqlc"
	nsqModel "bgs/nsqp/model"
	retryRunner "bgs/nsqp/runners/retry"
	"bgs/util"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"github.com/nsqio/go-nsq"
	"go.uber.org/zap"
	"strconv"
)

// RetryNsqFailHandler 触发重试nsq消息 handler
type RetryNsqFailHandler struct {
	dao         *db.PgDao
	slog        *zap.SugaredLogger
	retryRunner *retryRunner.NsqRetryRunner
}

// NewRetryNsqFailHandler create a new handler for RetryNsqFailNotification
func NewRetryNsqFailHandler(
	pgdao *db.PgDao,
	slog *zap.SugaredLogger,
	retryRunner *retryRunner.NsqRetryRunner,
) *RetryNsqFailHandler {
	return &RetryNsqFailHandler{
		dao:         pgdao,
		slog:        slog.With("module", "RetryNsqFailHandler"),
		retryRunner: retryRunner,
	}
}

// HandleMessage implements the Handler interface.
func (h *RetryNsqFailHandler) HandleMessage(m *nsq.Message) error {
	if len(m.Body) == 0 {
		// Returning nil will automatically send a FIN command to NSQ to mark the message as processed.
		return nil
	}

	// h.slog.Infof("RetryNsqFailMessage=>%s", string(m.Body))

	v := &nsqModel.CronMomentMessage{}
	if decodeErr := json.NewDecoder(bytes.NewReader(m.Body)).Decode(v); decodeErr != nil {
		return fmt.Errorf("无法解析信息[RetryNsqFailMessage]: %v", decodeErr)
	}
	ts := v.TS
	limit := util.Int32IfThenElse(v.Data.Limit > 0, v.Data.Limit, 100)
	reqID := strconv.FormatInt(ts, 10)
	ctx := context.WithValue(context.Background(), util.RequestID{}, reqID)

	err := h.dao.NewTxRunner().SqlcExecTx(ctx, func(q *sqlc.Queries) (err error) {
		nsqFails, err := q.ListNsqFailsByRetryCount(ctx, limit)
		if err != nil {
			return
		}
		for _, nsqFail := range nsqFails {
			err = h.retryRunner.PublishNsqFail(&nsqFail)
			if err != nil {
				if !nsqFail.IsSub {
					sqlErr := q.UpdateNsqFailRetryCount(ctx, sqlc.UpdateNsqFailRetryCountParams{
						ID:     nsqFail.ID,
						ErrMsg: fmt.Sprintf("pub nsq retry err:%s", err.Error()),
					})
					if sqlErr != nil {
						h.slog.Errorf("UpdateNsqFailRetryCount err===>%s", sqlErr.Error())
					}
				}
				err = nil
			} else {
				// 没有错误就删除nsq_fail
				h.slog.Infof("没有错误就删除当前的nsq_fail(%d)", nsqFail.ID)
				sqlErr := q.DeleteNsqFailByID(ctx, nsqFail.ID)
				if sqlErr != nil {
					h.slog.Errorf("DeleteNsqFailByID err===>%s", sqlErr.Error())
				}
			}
		}
		h.slog.Infof("a RetryNsqFailHandler-retry(%d) ok", len(nsqFails))
		return
	})

	if err != nil {
		return err
	}
	// Returning a non-nil error will automatically send a REQ command to NSQ to re-queue the message.
	return nil
}
